package com.niit.service

import com.niit.bean.Answer
import com.niit.utils.SparkUtils
import org.apache.spark.ml.recommendation.ALSModel
import org.apache.spark.sql.{DataFrame, SaveMode}
import org.apache.spark.streaming.dstream.DStream

class EDURecommendService {

  val spark = SparkUtils.takeSpark()
  import spark.implicits._
  import org.apache.spark.sql.functions._

  def dataAnalysis(answer:DStream[Answer]): Unit ={

    //DStream 转换rdd ==> DataFrame 、Dataset
    answer.foreachRDD(rdd=>{
      //1.加载模型
      val model_path = "output/als_model/1716975397774"
      val model: ALSModel = ALSModel.load(model_path)
      //2.切割实时数据中 studentid
      //2.1定义udf函数，将实时数据中的学生ID_45 切割出来 数据为45
      val id2Int = udf((student_id:String)=>{
          student_id.split("_")(1).toInt
      })
      //3.将rdd 转换成 DataFrame，因为ALS模型传入的参数必须DataFrame
      val answerDF: DataFrame = rdd.toDF()
        //由于是针对学生id进行错题的推荐，所以要将学生id切割出来
      val studentIdDF: DataFrame = answerDF.select(id2Int('student_id) as "student_id")

      //4.使用模型给学生推荐错题    要推荐的学生id     推荐10道错题
      val recommendDF: DataFrame = model.recommendForUserSubset(studentIdDF, 10)
      //4.1打印 推荐结果的数据结构 和内容
      recommendDF.printSchema()
      recommendDF.show(false)

      //5.处理推荐结果，取出学生id和题目id，将题目拼成一个字符串
      val res: DataFrame = recommendDF.as[(Int, Array[(Int, Float)])].map(t => {
        //模型中存储都是数字，后面要存数据库了，将学生 和 题目 进行拼接
        val student_id = "学生ID_" + t._1
        val question_id = t._2.map("题目ID_" + _._1).mkString(",")
        (student_id, question_id)
      }) toDF("student_id", "recommendations")

      //6.将answerDF 和 推荐结果 根据 student_id 进行 相关
      val allInfoDF: DataFrame = answerDF.join(res, "student_id")

      //7.将关联结果保存进数据库
      allInfoDF.write
        .format("jdbc")
        .option("url","jdbc:mysql://node1:3306/BD2?useUnicode=true&characterEncoding=utf8")
        .option("driver","com.mysql.jdbc.Driver")
        .option("user","root")
        .option("password","Niit@123")
        .option("dbtable","edu")
        .mode(SaveMode.Append)
        .save()
    })





  }

}
